package cn.monkey.transport.core;

import cn.monkey.transport.core.server.FilterChain;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @param <C>
 * @param <PKG>
 */
public abstract class InboundReceivedExchangeFactory<C, EXCHANGE extends Exchange, PKG> implements ExchangeFactory<C, EXCHANGE> {

    private static final Logger log = LoggerFactory.getLogger(InboundReceivedExchangeFactory.class);
    protected final FilterChain<? super EXCHANGE, PKG> filterChain;
    protected final Class<PKG> pkgType;
    protected final Scheduler scheduler;

    public InboundReceivedExchangeFactory(FilterChain<? super EXCHANGE, PKG> filterChain,
                                          Class<PKG> pkgType) {
        this(filterChain, pkgType, null);
    }

    public InboundReceivedExchangeFactory(FilterChain<? super EXCHANGE, PKG> filterChain,
                                          Class<PKG> pkgType,
                                          Scheduler scheduler) {
        Objects.requireNonNull(filterChain);
        Objects.requireNonNull(pkgType);
        this.filterChain = filterChain;
        this.pkgType = pkgType;
        this.scheduler = scheduler;
    }


    @Override
    @SuppressWarnings("unchecked")
    public EXCHANGE apply(String key, Exchange exchange, C c) {
        if (exchange == null) {
            throw new NullPointerException();
        }
        Flux<PKG> pkgFlux;
        if (exchange instanceof DelegatesExchange delegatesExchange) {
            List<Flux<PKG>> fluxList = delegatesExchange.delegates().stream().map(e -> e.inbound().receive(this.pkgType)).collect(Collectors.toList());
            pkgFlux = Flux.merge(fluxList);
        } else {
            pkgFlux = exchange.inbound().receive(this.pkgType);
        }
        EXCHANGE e = (EXCHANGE) exchange;
        Flux<PKG> flux = pkgFlux.doOnError(t -> log.error("message type: {} subscribe error:\n", this.pkgType.getName(), t));
        if (this.scheduler != null) {
            flux.subscribeOn(this.scheduler);
            exchange.setAttribute(Scheduler.class.getName(), this.scheduler);
        }
        flux.subscribe(pkg -> this.filterChain.doFilter(e, pkg));
        return e;
    }
}
